package com.shujia.stream

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object Demo2UpdateStateByKey {
  def main(args: Array[String]): Unit = {

    //1、创建spark 环境
    var conf = new SparkConf()
      .setMaster("local[2]")
      .setAppName("wc")

    val sc = new SparkContext(conf)


    //2、创建SparkStreaming的上下文对象
    /**
      * 需要指定batch的间隔时间，多久计算一次的意思
      * 时间越长延迟就越长
      * 时间越吞吐量会降低
      *
      */
    val ssc = new StreamingContext(sc, Durations.seconds(5))

    /**
      * 指定checkpoint ，保持计算的状态，
      * 可以是hdfs的路径
      *
      */

    ssc.checkpoint("spark/data/checkpoint")

    //3、读取数据
    //构建Dstream,  被动接受数据
    val ds: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.129.101", 8888)


    //处理数据
    val kvDS: DStream[(String, Int)] = ds.flatMap(_.split(",")).map((_, 1))

    /**
      * seq: 当前batch一个key所有的value
      * option: 之前所有batch计算的状态，如果是第一次计算，就为空
      *
      *
      * 使用当前batch的数据去更新之前的状态，返回一个新的状态
      */

    val fun = (seq: Seq[Int], option: Option[Int]) => {
      //当前batch一个单词的数量
      val sum = seq.sum

      //之前统计的单词的数量
      val last = option.getOrElse(0)

      //返回总的数量
      Option(sum + last)
    }

    /**
      * 有状态算子
      *
      * 需要设置checkpoint，保存计算的状态
      */

    val countDS: DStream[(String, Int)] = kvDS.updateStateByKey(fun)


    countDS.print()


    //启动任务
    ssc.start()
    //等待关闭
    ssc.awaitTermination()

  }
}
